package net.bwie.realtime.jtp.dwd.log.function;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.JSONArray;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

public class AppLogSplitProcessFunction2 extends ProcessFunction<String,String> {
    private final OutputTag<String> errorTag;
    private final OutputTag<String> startTag;
    private final OutputTag<String> displayTag;
    private final OutputTag<String> actionTag;

    public AppLogSplitProcessFunction2(OutputTag<String> errorTag, OutputTag<String> startTag,
                                       OutputTag<String> displaysTag, OutputTag<String> actionTag) {
        this.errorTag = errorTag;
        this.startTag = startTag;
        this.displayTag = displaysTag;
        this.actionTag = actionTag;
    }


    @Override
    public void processElement(String value, Context ctx, Collector<String> out) throws Exception {
        //第一步：解析JSON数据
        JSONObject jsonObject = JSON.parseObject(value);

        //第二步：错误日志
        if(null!=jsonObject.get("err")){
            //侧边流输出
            ctx.output(errorTag,value);
            //错误数据已经单独保存，无需在保存，直接删除
            jsonObject.remove("err");
        }

        //第三步：启动日志
        if(null!=jsonObject.get("start")){
            //侧边流输出
            ctx.output(startTag,jsonObject.toJSONString());
        }
        //第四步：页面日志
        else {
            //获取曝光和动作数据
            JSONArray displaysJSONArray = jsonObject.getJSONArray("displays");
            JSONArray actionsJSONArray = jsonObject.getJSONArray("actions");

            //删除曝光和动作
            jsonObject.remove("displays");
            jsonObject.remove("actions");

            //输出页面日志
            out.collect(jsonObject.toJSONString());

            //第五步：曝光日志
            if(null!=displaysJSONArray && displaysJSONArray.isEmpty()){
                for (int i=0;i < displaysJSONArray.size();i++){
                    JSONObject displaysJSONObject = displaysJSONArray.getJSONObject(i);
                    //添加字段值
                    jsonObject.put("displays",displaysJSONObject);
                    //输出
                    ctx.output(displayTag,jsonObject.toJSONString());
                }
            }
            //将上述添加曝光数据字段值删除
            jsonObject.remove("displays");

            //第六步：动作日志
            if(null!=actionsJSONArray && actionsJSONArray.isEmpty()){
                for (int i=0;i<actionsJSONArray.size();i++){
                    JSONObject actionJSONObject = actionsJSONArray.getJSONObject(i);
                    //添加字段值
                    jsonObject.put("action",actionJSONObject);
                    //输出
                    ctx.output(actionTag,jsonObject.toJSONString());
                }
            }
        }
    }
}
